﻿#include "threadpool.h"

#include <stdlib.h>
#include <assert.h>


#ifdef _WIN32 //为低版本window兼容而写

#define COMPATIBLE_XP  1//是否兼容XP标志

#include <process.h>
#include <Windows.h>

typedef struct __thread_cond
{
    //条件变量内部的区间代码需要互斥，所以需要一个临界区
    CRITICAL_SECTION hLock;

    //目前有多少条线程陷入了等待，条件变量陷入等待会递增uTotalWait
    //条件变量被唤醒会递减 uTotalWait
    LONG nTotalWait;

    //uWaitCount 表示应该剩余多少条线程进行等待，
    //调用唤醒函数时，此值会减少。调用等待函数时，此值会增加
    //uWaitCount 与 uTotalWait 之间的差，代表需要唤醒的线程条数
    //通过hWaitEvent唤醒线程，使uTotalWait递减到与uWaitCount数量一致
    LONG nWaitCount;

    //可能会等待的数量，这个是用来加速的，如果没有这个变量，那么每次调用
    //唤醒函数时，即时没有线程在等待，也会陷入临界区抢锁的开销，有了这个
    //原子变量，那么我们可以在唤醒前判断一下是否有线程调用了 wait 函数，
    //有的话，再进行唤醒，没有的话直接返回就可以了
    //这个变量的运算，是与 nWaitCount 同步的
    LONG nMaybeCount;

    //负责等待与唤醒线程
    HANDLE hWaitEvent;
} thread_cond_t;

//初始化函数，负责进行简单的赋值
static BOOL pthread_cond_init(thread_cond_t* cond_t)
{
    memset(cond_t, 0, sizeof(thread_cond_t));

    //无信号，且自动重置信号，也就是每次只能唤醒一条线程
    cond_t->hWaitEvent = CreateEvent(NULL, FALSE, FALSE, NULL);

    if (cond_t->hWaitEvent)
    {
        InitializeCriticalSection(&cond_t->hLock);
    }

    return !!cond_t->hWaitEvent;
}

//销毁条件变量对象
static void pthread_cond_destroy(thread_cond_t* cond_t)
{
    if (cond_t->hWaitEvent)
    {
        DeleteCriticalSection(&cond_t->hLock);
        CloseHandle(cond_t->hWaitEvent);
    }

    memset(cond_t, 0, sizeof(thread_cond_t));
}

//唤醒一个正在等待的条件变量
static int pthread_cond_signal(thread_cond_t* cond_t)
{
    int nRet = -1;

    //if(0 == cond_t->nMaybeCount)
    if (NULL == cond_t->hWaitEvent)
    {
        SetLastError(ERROR_INVALID_HANDLE);
        return -1;
    }

    if (0 == InterlockedCompareExchange(&cond_t->nMaybeCount, 0, 0))
    {
        //没有线程进入等待函数
        return 0;
    }

    EnterCriticalSection(&cond_t->hLock);

    nRet = cond_t->nWaitCount;
    if (cond_t->nWaitCount > 0)
    {
        if (TRUE == SetEvent(cond_t->hWaitEvent))
        {
            InterlockedDecrement(&cond_t->nMaybeCount);
            cond_t->nWaitCount--;
        }
        else
        {
            nRet = -1;
        }
    }

    LeaveCriticalSection(&cond_t->hLock);
    return nRet;
}

//唤醒所有的正在等待的条件变量
static int pthread_cond_broadcast(thread_cond_t* cond_t)
{
    int nRet = -1;
    //if(0 == cond_t->nMaybeCount)

    if (NULL == cond_t->hWaitEvent)
    {
        SetLastError(ERROR_INVALID_HANDLE);
        return -1;
    }

    if (0 == InterlockedCompareExchange(&cond_t->nMaybeCount, 0, 0))
    {
        //没有线程进入等待函数
        return 0;
    }

    EnterCriticalSection(&cond_t->hLock);

    nRet = cond_t->nWaitCount;
    if (cond_t->nWaitCount > 0)
    {
        if (TRUE == SetEvent(cond_t->hWaitEvent))
        {
            InterlockedExchange(&cond_t->nMaybeCount, 0);
            cond_t->nWaitCount = 0;
        }
        else
        {
            nRet = -1;
        }
    }

    LeaveCriticalSection(&cond_t->hLock);
    return nRet;
}

//让条件变量陷入等待，条件变量陷入等待会释放原来的互斥量，
static DWORD pthread_cond_timedwait(CRITICAL_SECTION* lock, thread_cond_t* cond_t, DWORD tvwait)
{
    DWORD dwError = 0;
    BOOL bWake = FALSE;
    DWORD dwTimeout = tvwait;
    DWORD dwEnterTime = GetTickCount();

    if (NULL == cond_t->hWaitEvent)
    {
        return ERROR_INVALID_HANDLE;
    }

    if (0 == tvwait)
    {
        //放弃一次时间片：
        //Sleep(0);  

        //放弃一次锁权限 ：
        //LeaveCriticalSection(&cond_t->hLock);
        //EnterCriticalSection(&cond_t->hLock);
        return ERROR_TIMEOUT;
    }

    EnterCriticalSection(&cond_t->hLock);
    LeaveCriticalSection(lock);

    if (THREADPOOL_LONG_MAX == cond_t->nTotalWait)
    {
        dwError = ERROR_IS_JOIN_PATH; //资源不足
        goto FINISH;
    }

    cond_t->nTotalWait++;
    cond_t->nWaitCount++;
    InterlockedIncrement(&cond_t->nMaybeCount);

    do
    {
        LeaveCriticalSection(&cond_t->hLock);

        if (-1 != tvwait)
        {
            dwTimeout = GetTickCount() - dwEnterTime; //先获取已经消耗掉的时间

            //如果超时时间大于已经消耗的时间，那么计算时间差继续等待，
            //否则超时时间为 0
            dwTimeout = tvwait > dwTimeout ? tvwait - dwTimeout : 0;
        }

        dwError = WaitForSingleObjectEx(cond_t->hWaitEvent, dwTimeout, FALSE);
        EnterCriticalSection(&cond_t->hLock);

        switch (dwError)
        {
        case WAIT_OBJECT_0:
            {
                dwError = 0; //必须确保错误码就是0

#if 1 /*允许虚假唤醒*/

                bWake = TRUE; //允许虚假唤醒
                //其实最多只有 nWaitCount 等于 nTotalWait的情况
                //nWaitCount 是不可能大于 nTotalWait，
                //assert(cond_t->nTotalWait >= cond_t->nWaitCount)
                if (cond_t->nWaitCount == cond_t->nTotalWait)
                {
                    dwError = ERROR_BAD_ENVIRONMENT; //环境错误

                    //因为nTotalWait此时还没递减，所以nTotalWait必然大于等于 1
                    //间接地证明了 nWaitCount 也必然大于等于 1
                    InterlockedDecrement(&cond_t->nMaybeCount);
                    cond_t->nWaitCount--;
                }

#else /*不允许虚假唤醒*/
            
            if (cond_t->nTotalWait > cond_t->nWaitCount)//判断一下是否符合唤醒的条件
            {
                bWake = TRUE;
            }
            /*else
            * {
            *     bWake = FALSE; //虚假唤醒
            * }
            */
#endif
            }
            break;
        case WAIT_TIMEOUT:
            {
                dwError = ERROR_TIMEOUT;
            }
            break;
        default:
            {
                dwError = GetLastError();
                if (0 == dwError)
                {
                    //必须强行给他一个错误吗，错误码为空回令接下来的判断不正确
                    //按道理是一定不会进入这个判断的
                    dwError = ERROR_UNKNOWN_PROPERTY; //未知错误
                }
            }
            break;
        }
    }
    while (FALSE == bWake && 0 == dwError);

    //dwError 不为 0，可能不是被唤醒函数唤醒的
    //不是调用唤醒函数唤醒的情况，要主动将 nWaitCount 减 1
    //因为本次函数的醒来，不属于唤醒函数的功劳，
    if (FALSE == bWake && dwError && cond_t->nWaitCount > 0)
    {
        InterlockedDecrement(&cond_t->nMaybeCount);
        cond_t->nWaitCount--;
    }

    cond_t->nTotalWait--;

    //如果nTotalWait大于nWaitCount，说明还有其他线程需要被唤醒
    if (cond_t->nTotalWait > cond_t->nWaitCount)
    {
        if (FALSE == SetEvent(cond_t->hWaitEvent))
        {
            dwError = GetLastError();
        }
    }

FINISH:

    //先把锁释放了，不然会产生死锁
    LeaveCriticalSection(&cond_t->hLock);

    EnterCriticalSection(lock);
    return dwError;
}

#endif//为低版本window兼容而写


#define FOREVER  -1

#define threadpool_malloc(l) malloc(l)
#define threadpool_realooc(p, l) realloc(p, l)
#define threadpool_free(p) free(p)
typedef long long _time_t;

#define threadpool_wait_cond(pool, cv, tival) threadpool_wait_cond_((pool), &(cv), (tival))

#ifdef _WIN32

#define EWAITTIMEOUT  ERROR_TIMEOUT


#if COMPATIBLE_XP //是否兼容低版本windows

typedef unsigned int THREADRET;
typedef HANDLE _thread_t;
typedef thread_cond_t _thread_cond_t;
typedef CRITICAL_SECTION _thread_mutex_t;

#define threadpool_lock(pool) EnterCriticalSection(&(pool)->lock)
#define threadpool_unlock(pool) LeaveCriticalSection(&(pool)->lock)
#define threadpool_wake_one(cv) pthread_cond_signal(&(cv))
#define threadpool_wake_all(cv) pthread_cond_broadcast(&(cv))
#define threadpool_init_cond(cv) pthread_cond_init(&(cv))
#define threadpool_destroy_cond(cv) pthread_cond_destroy(&(cv))

#else //是否兼容低版本windows

typedef void*                    THREADRET;
typedef HANDLE                   _thread_t;
typedef CONDITION_VARIABLE       _thread_cond_t;
typedef CRITICAL_SECTION         _thread_mutex_t;

#define threadpool_lock(pool) EnterCriticalSection(&(pool)->lock)
#define threadpool_unlock(pool) LeaveCriticalSection(&(pool)->lock)
#define threadpool_wake_one(cv) WakeConditionVariable(&(cv))
#define threadpool_wake_all(cv) WakeAllConditionVariable(&(cv))
#define threadpool_init_cond(cv) InitializeConditionVariable(&(cv))
#define threadpool_destroy_cond(cv) 

#endif

#define get_current_thread_id  GetCurrentThreadId


#else
#include <stdlib.h>
#include <pthread.h>
#include <errno.h>
#include <sys/time.h>
#define max(a,b)    (((a) > (b)) ? (a) : (b))
#define EWAITTIMEOUT  ETIMEDOUT

typedef pthread_t                   _thread_t;
typedef pthread_cond_t       _thread_cond_t;
typedef pthread_mutex_t         _thread_mutex_t;

#define threadpool_lock(pool) pthread_mutex_lock(&(pool)->lock)
#define threadpool_unlock(pool) pthread_mutex_unlock(&(pool)->lock)
#define threadpool_wake_one(cv) pthread_cond_signal(&(cv))
#define threadpool_wake_all(cv) pthread_cond_broadcast(&(cv))
#define threadpool_init_cond(cv) pthread_cond_init(&(cv), NULL)
#define threadpool_destroy_cond(cv) pthread_cond_destroy(&(cv))

#define get_current_thread_id  pthread_self

#endif

typedef struct __HEAPELEMENT
{
    _time_t tiValue;
    struct __POOLTIMER* pTimer;
} HEAPELEMENT;

typedef struct __MINHEAP
{
    thread_ulong_t dwIncrement; //空间不足时,每次递增多少空间
    thread_ulong_t dwCount; //元素数量
    HEAPELEMENT* lpArrayElement;
} MINHEAP;

typedef struct __JOB
{
    thread_bool_t priority; //是否优先处理
    queue_hook_cb callback; //回调
    void* arg; //参数
    thread_ulong_t index; //内存索引标记
    struct __JOB* next;
} JOB, *PJOB;

typedef struct __JOB_QUEUEHEAD
{
    struct __JOB* queue_frist; //队列中第一个消息
    struct __JOB* queue_last; //队列中最后一个消息
} JOB_QUEUEHEAD;

typedef struct __POOLTIMER
{
    thread_ulong_t index; //小根堆的位置
    thread_ulong_t ref; //引用计数

    thread_ulong_t threadid;

    //需要一个事件(HEVENT)或者条件变量来保证数据的线程安全,因为当timer执行时,会跳出锁
    //如果此时正好释放定时器的话,就会引起崩溃,所以当定时器开始时,先设置HEVENT
    //为无信号, 定时器执行完毕后,设置为有信号
    //void * signal;
    _thread_cond_t cv_kill;
    _time_t timeout; //毫秒
    timer_hook_cb callback;
    long event;
    void* arg;
} POOLTIMER;

struct __THREADPOOL
{
    int thread_num; //常驻线程数
    int thread_max_num; //最大线程数
    int thread_cur_num; //当前线程数
    int queue_cur_num; //当前队列消息数
    int queue_max_num; //最大队列消息数
    thread_bool_t pool_close; //正在关闭线程池
    _thread_t* pthreads; //线程句柄数组
    thread_bool_t block_full; //默认TRUE 队列满时,阻塞等待. FALSE 失败退出

    _thread_mutex_t lock; //消息读写锁
    _thread_cond_t cv_not_empty; //队列消息非空
    _thread_cond_t cv_not_full; //队列消息未满
    _thread_cond_t cv_destroy; //注销通知
    _thread_cond_t cv_create_thread; //等待线程创建完成

    //struct __JOB * queue_frist;//队列中第一个消息
    //struct __JOB * queue_last;//队列中最后一个消息
    //struct __JOB * queue_priority_last;//优先队列中的最后一个消息位置,具有优先属性的消息,将从queue_frist开始插入

    unsigned int priority_count; //任务优先级个数
    struct __JOB_QUEUEHEAD* job_queue_head; //任务队列数组

    JOB* job_mempool; //工作内存池,避免重复申请内存
    struct __JOB* cache_frist; //缓存链第一个

    MINHEAP* time_heap; //timer小根堆

    unsigned int max_timer_count; //最大允许创建的定时器数量
    unsigned int timer_cur_count; //定时器数量
    unsigned int timer_array_count; //线程中为timer提供的缓冲大小
    POOLTIMER* thread_timer; //最多定义N个timer

    queue_hook_cb stable_cb; //固定的回调,保留
};

static void empty(void* p, int size)
{
    while (size--)
    {
        ((char*)p)[size] = '\0';
    }
}

static _time_t get_time()
{
    _time_t ti = 0;
#ifdef _WIN32

    QueryPerformanceCounter((LARGE_INTEGER*)&ti);
    ti = ti / 1000;

#else

    struct timeval tival = { 0 };
    gettimeofday(&tival, NULL);

    ti += tival.tv_sec * 1000;
    ti += tival.tv_usec / 1000;
#endif
    return ti;
}


/*
下面是timer
*/

#define MHELEM_SET_INDEX(pElem, Value)  ((pElem)->pTimer->index = Value)
#define MHELEM_UPDATE_INDEX(lpElemArray, Index)  ((lpElemArray)[Index].pTimer->index = (Index))
#define MHELEM_INVALID_INDEX(pElem)  (-1 == (pElem)->pTimer->index ? 1 : 0)

static MINHEAP* min_heap_new_(thread_ulong_t dwIncrement)
{
    if (0 == dwIncrement) return NULL;
    MINHEAP* pMinHeap = threadpool_malloc(sizeof(MINHEAP));
    empty(pMinHeap, sizeof(MINHEAP));
    pMinHeap->dwIncrement = dwIncrement;

    pMinHeap->lpArrayElement = (HEAPELEMENT*)threadpool_malloc(sizeof(HEAPELEMENT) * dwIncrement);

    if (NULL == pMinHeap->lpArrayElement)
    {
        threadpool_free(pMinHeap);
        pMinHeap = NULL;
    }
    return pMinHeap;
}

static void min_heap_free_(MINHEAP* pMinHeap)
{
    if (pMinHeap->lpArrayElement)
    {
        threadpool_free(pMinHeap->lpArrayElement);
    }
    threadpool_free(pMinHeap);
}

static __inline thread_ulong_t min_heap_get_count_(MINHEAP* pMinHeap)
{
    return pMinHeap->dwCount;
}

static void min_heap_clear_(MINHEAP* pMinHeap)
{
    if (pMinHeap->dwCount > pMinHeap->dwIncrement)
    {
        //需要缩小
        //由于是缩小, 这里是一定成功的
        pMinHeap->lpArrayElement = (HEAPELEMENT*)threadpool_realooc(pMinHeap->lpArrayElement,
                                                                    pMinHeap->dwIncrement * sizeof(HEAPELEMENT));
        assert(pMinHeap->lpArrayElement && "min_heap_clear 发生不可预料的情况!");
    }

    pMinHeap->dwCount = 0;
    empty(pMinHeap->lpArrayElement, pMinHeap->dwIncrement * sizeof(HEAPELEMENT));
}

static thread_bool_t min_heap_top_(MINHEAP* pMinHeap, HEAPELEMENT* pElem)
{
    if (NULL == pMinHeap || NULL == pElem)
    {
        assert(0 && "min_heap_top 传入无效的指针, 请检查!");
        return FALSE;
    }

    if (0 == pMinHeap->dwCount) return FALSE;

    *pElem = pMinHeap->lpArrayElement[0];

    return TRUE;
}

static thread_bool_t min_heap_push_(MINHEAP* pMinHeap, const HEAPELEMENT* pElem)
{
    thread_ulong_t dwBytes = 0;
    HEAPELEMENT* lpArray = NULL;
    thread_ulong_t nIndex = 0;
    thread_ulong_t j = 0;
    if (NULL == pMinHeap || NULL == pElem)
    {
        assert(0 && "min_heap_push 传入无效的指针, 请检查!");
        return FALSE;
    }

    if (FALSE == MHELEM_INVALID_INDEX(pElem)) //这个元素已经有效
    {
        assert(0 && "min_heap_push 元素可能已经存在于堆中, 请检查!");
        return FALSE;
    }

    if (pMinHeap->dwCount > 0 && 0 == pMinHeap->dwCount % pMinHeap->dwIncrement)
    {
        //内存刚好够用, 现在还要插入一个, 那就需要重新分配空间
        dwBytes = sizeof(HEAPELEMENT) * (pMinHeap->dwCount + pMinHeap->dwIncrement);
        if (NULL == (lpArray = (HEAPELEMENT*)threadpool_realooc(pMinHeap->lpArrayElement, dwBytes)))
        {
            return FALSE;
        }

        pMinHeap->lpArrayElement = lpArray;
        empty(pMinHeap->lpArrayElement + pMinHeap->dwCount, sizeof(HEAPELEMENT) * pMinHeap->dwIncrement);
    }

    //内存问题已处理好, 下面开始正式插入

    lpArray = pMinHeap->lpArrayElement;
    nIndex = pMinHeap->dwCount; //nIndex指向待调整元素的位置，即其数组下标，初始指向新元素所在的堆尾位置

    lpArray[pMinHeap->dwCount++] = *pElem; //向堆尾添加新元素, 并增加数组长度

    while (0 != nIndex)
    {
        j = (nIndex - 1) / 2; //j指向下标为nIndex的元素的双亲
        if (pElem->tiValue >= lpArray[j].tiValue) //若新元素大于待调整元素的双亲，则比较调整结束，退出循环
            break;

        lpArray[nIndex] = lpArray[j]; //将双亲元素下移到待调整元素的位置

        MHELEM_UPDATE_INDEX(lpArray, nIndex);
        nIndex = j; //使待调整位置变为其双亲位置，进行下一次循环
    }
    lpArray[nIndex] = *pElem; //把新元素调整到最终位置
    MHELEM_UPDATE_INDEX(lpArray, nIndex);

    return TRUE;
}

static thread_bool_t min_heap_erase_(MINHEAP* pMinHeap, thread_ulong_t uIndex)
{
    thread_ulong_t dwBytes = 0;
    HEAPELEMENT* lpArray = NULL;
    //unsigned long uIndex = 0;//用nIndex指向待调整元素的位置，初始指向堆顶位置
    unsigned long j = 0; //j是默认指向左边, 如果左边比右边大,那么指向右边
    HEAPELEMENT Element;
    thread_bool_t bReAlloc = FALSE;

    if (NULL == pMinHeap)
    {
        assert(0 && "min_heap_erase 传入无效的指针, 请检查!");
        return FALSE;
    }

    if (uIndex > pMinHeap->dwCount) return FALSE;

    lpArray = pMinHeap->lpArrayElement;
    MHELEM_SET_INDEX(&lpArray[uIndex], -1); //已经被移除

    if (0 == --pMinHeap->dwCount) return TRUE;

    Element = lpArray[pMinHeap->dwCount]; //将待调整的原堆尾元素暂存temp中，以便放入最终位置
    j = 2 * uIndex + 1; //用j指向nIndex的左孩子位置，初始指向下标为1的位置

    while (pMinHeap->dwCount - 1 >= j) //寻找待调整元素的最终位置，每次使孩子元素上移一层，调整到孩子为空时止
    {
        //若存在右孩子且较小，使j指向右孩子
        if (pMinHeap->dwCount - 1 > j && lpArray[j].tiValue > lpArray[j + 1].tiValue) //左比右大
            j++; //指向右边, 指向小的位置

        if (lpArray[j].tiValue >= Element.tiValue) //若temp比其较小的孩子还小，则调整结束，退出循环
            break;

        lpArray[uIndex] = lpArray[j]; //否则，将孩子元素移到双亲位置
        MHELEM_UPDATE_INDEX(lpArray, uIndex);
        uIndex = j; //将待调整位置变为其较小的孩子位置
        j = 2 * uIndex + 1; //将j变为新的待调整位置的左孩子位置，继续下一次循环
    }

    lpArray[uIndex] = Element;
    MHELEM_UPDATE_INDEX(lpArray, uIndex);

    if (pMinHeap->dwCount >= pMinHeap->dwIncrement && 0 == pMinHeap->dwCount % pMinHeap->dwIncrement)
    {
        //此时内存刚好多出一个, 现在还移除一个, 那就重新分配空间, 将空间压缩到最小
        dwBytes = sizeof(HEAPELEMENT) * (pMinHeap->dwCount);

        //因为是缩小空间, 所以realloc是不会失败的
        pMinHeap->lpArrayElement = (HEAPELEMENT*)threadpool_realooc(pMinHeap->lpArrayElement, dwBytes);

        assert(lpArray == pMinHeap->lpArrayElement && "min_heap_erase 发生不可预料的情况!");
    }

    return TRUE;
}

static thread_bool_t min_heap_pop_(MINHEAP* pMinHeap, HEAPELEMENT* pElem)
{
    if (TRUE == min_heap_top_(pMinHeap, pElem))
    {
        return min_heap_erase_(pMinHeap, 0);
    }

    return FALSE;
}

static int min_heap_popbat_(MINHEAP* pMinHeap, _time_t tiCutValue, HEAPELEMENT* pElem, int nCount)
{
    int nIndex = 0;
    if (NULL == pMinHeap || NULL == pElem)
    {
        assert(0 && "min_heap_popbat 传入无效的指针, 请检查!");
        return -1;
    }

    if (0 == pMinHeap->dwCount) return 0;

    //将符合触发条件的event都取出来
    for (nIndex = 0; nCount > nIndex; nIndex++)
    {
        if (pMinHeap->dwCount > 0 && tiCutValue >= pMinHeap->lpArrayElement[0].tiValue)
        {
            min_heap_pop_(pMinHeap, &pElem[nIndex]);
        }
        else
        {
            break;
        }
    }

    return nIndex;
}


/*
上面是timer的相关算法, 下面是线程池的执行流程
*/

static thread_bool_t threadpool_active(THREADPOOL pool)
{
    if (pool)
    {
        //如果没有任务，在外部设置定时器后，将无法进行第一次唤醒，所以激活一下，使定时器生效
        //其实一般不太建议使用该函数，当然你也可以在loop前设置定时器，这样就不需要该函数了
        threadpool_wake_one(pool->cv_not_empty);
        return TRUE;
    }
    return FALSE;
}

static int threadpool_reset_timer_nolock(THREADPOOL pool, POOLTIMER* timer)
{
    int ret;
    HEAPELEMENT element = {0};

    if (TRUE == pool->pool_close)
    {
        return -1;
    }

    if (0 == timer->ref)
    {
        //定时器已经释放
        return 0;
    }

    if (-1 != timer->index)
    {
        //在回调线程释放timer后，其他线程重用了这个timer，就会发生这种情况
        return 0;
    }

    element.tiValue = timer->timeout + get_time();
    element.pTimer = timer;

    ret = min_heap_push_(pool->time_heap, &element);
    timer->threadid = 0;

    if (0 == timer->index)
    {
        threadpool_active(pool);
    }

    return ret;
}

static int threadpool_get_timer_nolock(THREADPOOL pool, HEAPELEMENT* timer_array, _time_t ti_cache,
                                       thread_ulong_t threadid)
{
    int count = 0;
    POOLTIMER* timer = NULL;
    int i = 0;

    if (0 == timer_array) return 0;

    empty(timer_array, sizeof(HEAPELEMENT) * pool->timer_array_count);
    count = min_heap_popbat_(pool->time_heap, ti_cache, timer_array, pool->timer_array_count);

    //设置当前调用线程
    for (i = 0; count > i; i++)
    {
        timer = timer_array[i].pTimer;
        timer->threadid = threadid;
    }

    return count;
}

static JOB* threadpool_alloc_job_nolock(THREADPOOL pool)
{
    JOB* job = NULL;

    assert(pool->cache_frist && "thread_pool 缓存队列不可能为空！");

    job = pool->cache_frist;
    pool->cache_frist = pool->cache_frist->next;

    return job;
}

static void threadpool_free_job_nolock(THREADPOOL pool, JOB* job)
{
    job->next = pool->cache_frist;
    pool->cache_frist = job;
}

static thread_ulong_t threadpool_wait_cond_(THREADPOOL pool, _thread_cond_t* cond_t, thread_ulong_t tvwait)
{
    thread_ulong_t error = 0;

    if (0 == tvwait) return EWAITTIMEOUT;
#ifndef _WIN32

    struct timespec abstime = { 0 };
    struct timeval tival = { 0 };

    if (FOREVER == tvwait)
    {
        error = pthread_cond_wait(cond_t, &pool->lock);
    }
    else
    {
        gettimeofday(&tival, NULL);
        abstime.tv_sec = tival.tv_sec + (tvwait) / 1000;
        if (0 == abstime.tv_sec)
        {
            //pthread_cond_timedwait至少需要休眠一秒, 否则返回EINVAL
            abstime.tv_sec = 1;
            abstime.tv_nsec = 0;
        }
        else
        {
            abstime.tv_nsec = (tival.tv_usec + (tvwait) % 1000) * 1000000;
        }

        error = pthread_cond_timedwait(cond_t, &pool->lock, &abstime);
}


#else

#if COMPATIBLE_XP

    error = pthread_cond_timedwait(&pool->lock, cond_t, tvwait);

#else
    if (FALSE == SleepConditionVariableCS(cond_t, &(pool)->lock, (tvwait)))
    {
        error = GetLastError();
    }
#endif

#endif
    return error;
}


static THREADRET TPSTDCALL threadpool_worker(void* arg)
{
    THREADPOOL pool = arg;
    unsigned int i = 0;
    JOB* job = NULL;
    POOLTIMER* timer = NULL;
    HEAPELEMENT* timer_array = NULL;

    int timer_count = 0; //触发的timer个数

    HEAPELEMENT element = {0};
    _time_t ti_cache = 0;
    thread_ulong_t timeout = FOREVER;
    thread_ulong_t last_error = 0;
    int handle_count = 0; //任务累计

    thread_ulong_t threadid = get_current_thread_id();
    thread_bool_t is_permanent_thread = TRUE; //是否常驻线程
    thread_bool_t all_quit = FALSE;

    threadpool_lock(pool);
    if (pool->thread_num)
    {
        is_permanent_thread = pool->thread_num >= pool->thread_cur_num ? TRUE : FALSE;
        threadpool_wake_one(pool->cv_create_thread);
    }
    threadpool_unlock(pool);

    if (is_permanent_thread && pool->timer_array_count > 0)
    {
        //常驻线程中才会触发定时器
        timer_array = (HEAPELEMENT*)threadpool_malloc(sizeof(HEAPELEMENT) * pool->timer_array_count);
        assert(timer_array && "threadpool_worker 致命的内存分配错误!");
    }

    while (TRUE)
    {
    BEGIN:
        timeout = FOREVER;
        timer_count = 0;

        threadpool_lock(pool);
        //通过iotime.key.data是否为NULL, 来得知上一个定时器是否被处理
        //被处理的话就提取一个新的

        if (timer_array && pool->timer_cur_count && handle_count > pool->thread_cur_num
            && pool->queue_cur_num && FALSE == pool->pool_close)
        {
            //定时器优先级很低，当任务很多的时候我们每处理大于线程数或以上的任务才行检查一次定时器
            if (TRUE == min_heap_top_(pool->time_heap, &element))
            {
                ti_cache = get_time();
                if (ti_cache >= element.tiValue) //时间到
                {
                    if (timer_count = threadpool_get_timer_nolock(pool, timer_array, ti_cache, threadid))
                    {
                        goto FINISH;
                    }
                }
            }
        }

        while (0 == pool->queue_cur_num)
        {
            if (FALSE == is_permanent_thread || TRUE == pool->pool_close)
            {
                pool->thread_cur_num--;
                if (0 == pool->thread_cur_num) all_quit = TRUE; //全都已经退出
                threadpool_unlock(pool);
                goto BYEBYE;
            }

            if (timer_array && TRUE == min_heap_top_(pool->time_heap, &element))
            {
                ti_cache = get_time();
                timeout = (thread_ulong_t)(element.tiValue > ti_cache ? element.tiValue - ti_cache : 0);
            }

            if (last_error = threadpool_wait_cond(pool, pool->cv_not_empty, timeout))
            {
                if (EWAITTIMEOUT == last_error && timer_array)
                {
                    ti_cache = get_time();
                    if (timer_count = threadpool_get_timer_nolock(pool, timer_array, ti_cache, threadid))
                    {
                        goto FINISH;
                    }
                }
                threadpool_unlock(pool);

                goto BEGIN;
            }
        }

        for (i = 0; pool->priority_count > i; i++) //按优先级取出一个任务
        {
            if (job = pool->job_queue_head[i].queue_frist)
            {
                pool->job_queue_head[i].queue_frist = job->next;

                if (NULL == pool->job_queue_head[i].queue_frist)
                {
                    //这个级别的任务被取完了,所以last也要置空
                    pool->job_queue_head[i].queue_last = NULL;
                }
                break;
            }
        }

        pool->queue_cur_num--;

        if (0 == pool->queue_cur_num)
        {
            //任务都执行完了， 再退出
            if (TRUE == pool->pool_close)
            {
                threadpool_wake_one(pool->cv_not_empty);
            }
        }
        else if ((pool->queue_cur_num + 1) == pool->queue_max_num)
        {
            threadpool_wake_one(pool->cv_not_empty);
            threadpool_wake_all(pool->cv_not_full);
        }
        else /*not full and not empty*/
        {
            threadpool_wake_one(pool->cv_not_empty); //通知另外一条线程
        }

    FINISH:
        threadpool_unlock(pool);

        if (timer_count)
        {
            //触发的是timer
            while (timer_count--) //反过来执行
            {
                timer = timer_array[timer_count].pTimer;

                //第一个参数要用arg，能兼容更多情况，比如this指针
                timer->callback(timer->arg, timer->event);

                //通过threadidId 或者 key 来判断对象是否已删除,这里用threadid判断最简单
                //如果timer被删除或重置为新的定时器, threadid可以是NULL, 也可以是其他线程, 
                //但绝不会是本线程, 因为本线程正在运行

                if (timer->threadid == threadid) //即使脏读也能判断正确
                {
                    threadpool_wake_one(timer->cv_kill);

                    threadpool_lock(pool);
                    threadpool_reset_timer_nolock(pool, timer);
                    threadpool_unlock(pool);
                }

                handle_count = 0; //重置
            }
        }
        else
        {
            //触发的是任务
            handle_count++; //任务累计
            job->callback(job->arg);
            threadpool_lock(pool);
            threadpool_free_job_nolock(pool, job);
            threadpool_unlock(pool);
        }
    } /*while(TRUE)*/

BYEBYE:

    if (timer_array)
        threadpool_free(timer_array);

    if (TRUE == all_quit) //全部都退出了
    {
        //这是最后一个线程
        threadpool_wake_one(pool->cv_destroy); //通知关闭
    }
    else
    {
        //还有线程未关闭
        threadpool_wake_one(pool->cv_not_empty);
    }

    return 0;
}


/*
下面是提供给外部程序的接口
*/

THREADPOOL_CONFIG* threadpool_config_new()
{
    THREADPOOL_CONFIG* config = threadpool_malloc(sizeof(THREADPOOL_CONFIG));

    if (NULL == config) return NULL;

    empty(config, sizeof(THREADPOOL_CONFIG));

    //以下提供一些默认值，用户可以在threadpool_create前将其修改掉
    config->thread_num = 1; //一条常驻线程
    config->max_thread_num = 1; //最多只有一条任务线程
    config->queue_max_num = 1024; //队列可容纳1024个任务
    config->max_timer_count = 32; //最大可创建32个定时器
    config->priority_count = 1; //只有一个优先级

    return config;
}

void threadpool_config_free(THREADPOOL_CONFIG* config)
{
    threadpool_free(config);
}

static void threadpool_release(THREADPOOL pool)
{
    if (pool->job_queue_head)
        threadpool_free(pool->job_queue_head);
    if (pool->pthreads)
        threadpool_free(pool->pthreads);
    if (pool->thread_timer)
        threadpool_free(pool->thread_timer);
    if (pool->time_heap) min_heap_free_(pool->time_heap);
    if (pool->job_mempool)
        threadpool_free(pool->job_mempool);
    if (pool)
        threadpool_free(pool);
}

THREADPOOL threadpool_create(THREADPOOL_CONFIG* config)
{
    THREADPOOL pool = NULL;
    int i = 0;
    unsigned int u = 0;

    if (NULL == config) return NULL;

    if (config->thread_num > (~(unsigned int)0 >> 1)
        || config->max_thread_num > (~(unsigned int)0 >> 1)
        || config->queue_max_num > (~(unsigned int)0 >> 1))
        return NULL;

    pool = (THREADPOOL)threadpool_malloc(sizeof(struct __THREADPOOL));
    if (NULL == pool) return NULL;
    empty(pool, sizeof(struct __THREADPOOL));

    pool->thread_num = config->thread_num;
    pool->thread_max_num = (0 == config->thread_num ? 0 : max(config->thread_num, config->max_thread_num));
    pool->thread_cur_num = 0;
    pool->queue_cur_num = 0;
    pool->queue_max_num = config->queue_max_num;
    pool->pool_close = FALSE;
    pool->priority_count = max(config->priority_count, 1);
    pool->max_timer_count = config->max_timer_count;

    pool->job_queue_head = (JOB_QUEUEHEAD*)threadpool_malloc(sizeof(JOB_QUEUEHEAD) * pool->priority_count);
    pool->job_mempool = (JOB*)threadpool_malloc(sizeof(JOB) * (pool->queue_max_num + pool->thread_max_num + 1));

    if (!pool->job_queue_head || !pool->job_mempool)
    {
        threadpool_release(pool);
        return NULL;
    }

    if (pool->max_timer_count > 0)
    {
        pool->thread_timer = (POOLTIMER*)threadpool_malloc(sizeof(POOLTIMER) * pool->max_timer_count);
        pool->time_heap = min_heap_new_(pool->max_timer_count);
        pool->timer_array_count = max(2, pool->max_timer_count/2);
        if (!pool->time_heap || !pool->thread_timer)
        {
            threadpool_release(pool);
            return NULL;
        }
    }

    empty(pool->job_queue_head, sizeof(JOB_QUEUEHEAD) * pool->priority_count);

    //加 1 是为了兼容queue方法, 因为queue的时候pool没有记录线程数, 
    //但是确实queue运行在某个线程, 并占用了一个内存空间
    empty(pool->job_mempool, sizeof(JOB) * (pool->queue_max_num + pool->thread_max_num + 1));

    int queue_count = pool->queue_max_num + pool->thread_max_num;
    pool->cache_frist = pool->job_mempool;
    for (i = 0; queue_count > i; i++)
    {
        pool->job_mempool[i].index = i;
        pool->job_mempool[i].next = &pool->job_mempool[i + 1];
    }

    empty(pool->thread_timer, sizeof(POOLTIMER) * pool->max_timer_count);

    for (u = 0; pool->max_timer_count > u; u++)
    {
        threadpool_init_cond(pool->thread_timer[u].cv_kill);
    }

    threadpool_init_cond(pool->cv_not_empty);
    threadpool_init_cond(pool->cv_not_full);
    threadpool_init_cond(pool->cv_destroy);
    threadpool_init_cond(pool->cv_create_thread);

#ifdef _WIN32

    InitializeCriticalSection(&pool->lock); //临界区

    if (pool->thread_num)
    {
        pool->pthreads = (_thread_t*)threadpool_malloc(sizeof(_thread_t) * pool->thread_num);

        for (i = 0; pool->thread_num > i; i++)
        {
            threadpool_lock(pool);
            //_beginthread会自动关闭句柄， ex不会

            pool->pthreads[i] = (_thread_t)_beginthreadex(NULL, 0, threadpool_worker, pool, 0, 0);

            if (INVALID_HANDLE_VALUE == pool->pthreads[i])
            {
                goto FAILED;
            }

            pool->thread_cur_num++;
            threadpool_wait_cond(pool, pool->cv_create_thread, FOREVER);
            threadpool_unlock(pool);
        }
    }

#else
    pthread_mutex_init(&pool->lock, NULL);
    
    if (pool->thread_num)
    {
        pool->pthreads = (_thread_t*)threadpool_malloc(sizeof(_thread_t) * pool->thread_num);

        for (i = 0; pool->thread_num > i; i++)
        {
            threadpool_lock(pool);
            if (0 == pthread_create(&pool->pthreads[i], NULL, threadpool_worker, (void *)pool))
            {
                pool->thread_cur_num++;
                threadpool_wait_cond(pool, pool->cv_create_thread, FOREVER);
            }
            else
            {
                goto FAILED;
            }

            threadpool_unlock(pool);
        }
    }

#endif

    //用自旋锁等待线程创建完成, 其实很快的
    return pool;

FAILED:
    threadpool_destroy(pool);
    return NULL;
}

thread_bool_t threadpool_destroy(THREADPOOL pool)
{
    threadpool_lock(pool);

    if (TRUE == pool->pool_close) //防止多次调用threadpool_destroy函数
    {
        threadpool_unlock(pool);
        return FALSE;
    }

    pool->pool_close = TRUE;

    threadpool_wake_one(pool->cv_not_empty);

    while (pool->thread_cur_num)
    {
        threadpool_wait_cond(pool, pool->cv_destroy, FOREVER);
    }

    threadpool_unlock(pool);

#ifdef _WIN32
    if (pool->thread_num)
    {
        //WaitForMultipleObjects(pool->thread_num, pool->pthreads, TRUE, FROEVER);
        for (int i = 0; pool->pthreads && pool->thread_num > i; i++)
        {
            CloseHandle(pool->pthreads[i]);
        }
    }

    DeleteCriticalSection(&pool->lock);
#else

    pthread_mutex_destroy(&pool->lock);
#endif

    threadpool_destroy_cond(pool->cv_not_empty);
    threadpool_destroy_cond(pool->cv_not_full);
    threadpool_destroy_cond(pool->cv_destroy);

    threadpool_release(pool);
    return TRUE;
}

int threadpool_add_job(THREADPOOL pool, int priority, thread_bool_t block_full,
                       queue_hook_cb callback, void* arg)
{
    int ret = 0;
    JOB* newjob = NULL;
    _thread_t tid = 0;
    JOB_QUEUEHEAD* job_queue_head = NULL; //任务队列数组

    if (-1 == priority) priority = pool->priority_count / 2;

    if (priority >= (int)pool->priority_count) return -1;

    if (NULL == callback) return -1;

    threadpool_lock(pool);

    if (FALSE == block_full && pool->queue_cur_num == pool->queue_max_num)
    {
        goto END;
    }

    if (FALSE == pool->pool_close
        && pool->queue_cur_num == pool->queue_max_num
        && pool->thread_max_num > pool->thread_cur_num)
    {
        //fprintf(stderr, "threadpool_add_job 任务数量已满[%d], 开启一个新线程", pool->queue_cur_num);
#ifdef _WIN32

        tid = (_thread_t)_beginthreadex(NULL, 0, threadpool_worker, pool, 0, 0);

        if (INVALID_HANDLE_VALUE == tid)
        {
            pool->thread_cur_num++;
            CloseHandle(tid);
            threadpool_wait_cond(pool, pool->cv_create_thread, FOREVER);
        }

#else

		if (0 == pthread_create(&tid, NULL, threadpool_worker, (void*)pool))
		{
			pool->thread_cur_num++;
			threadpool_wait_cond(pool, pool->cv_create_thread, FOREVER);
		}
#endif
    }

    while (FALSE == pool->pool_close && pool->queue_cur_num == pool->queue_max_num)
    {
        threadpool_wait_cond(pool, pool->cv_not_full, FOREVER);
    }

    if (TRUE == pool->pool_close)
    {
        ret = -1;
        goto END;
    }

    newjob = threadpool_alloc_job_nolock(pool);
    newjob->priority = priority;
    newjob->callback = callback;
    newjob->arg = arg;
    newjob->next = NULL;

    job_queue_head = &pool->job_queue_head[priority];
    if (job_queue_head->queue_frist == 0)
    {
        job_queue_head->queue_frist = newjob;
        job_queue_head->queue_last = newjob;

        if (0 == pool->queue_cur_num)
        {
            threadpool_wake_one(pool->cv_not_empty);
        }
    }
    else
    {
        job_queue_head->queue_last->next = newjob;
        job_queue_head->queue_last = newjob;
    }

    ret = ++pool->queue_cur_num;

END:
    threadpool_unlock(pool);
    return ret; //-1 失败，0队列满, 大于0成功返回任务插入的位置
}


THREADTIMER threadpool_set_timer(THREADPOOL pool, thread_ulong_t timeout, timer_hook_cb callback, void* arg, long event)
{
    POOLTIMER* timer = NULL;
    _thread_cond_t cv_kill = {0};
    int tid = 0;
    unsigned int i = 0;

    if (FOREVER == timeout)
    {
        assert(0 && "threadpool_set_timer 禁止设置用不超时的定时器");
        return tid;
    }

    threadpool_lock(pool);

    //因为线程池是不会频繁地增删定时器的,所以这里用for循环,不需要太高效的算法
    if (FALSE == pool->pool_close && pool->max_timer_count > pool->timer_cur_count)
    {
        for (i = 0; pool->max_timer_count > i; i++)
        {
            if (0 == pool->thread_timer[i].ref)
            {
                timer = &pool->thread_timer[i];
                pool->timer_cur_count++;
                tid = i + 1;
                break;
            }
        }
    }

    if (timer)
    {
        timer->timeout = timeout;
        timer->callback = callback;
        timer->event = event;
        timer->arg = arg;
        timer->ref = 1;
        timer->index = -1;
        threadpool_reset_timer_nolock(pool, timer);
    }

    threadpool_unlock(pool);

    return tid;
}

thread_bool_t threadpool_kill_timer(THREADPOOL pool, THREADTIMER tid)
{
    int ret = FALSE;
    _thread_cond_t cv_kill;
    POOLTIMER* timer = NULL;

    if (tid > pool->max_timer_count || 0 > tid) return FALSE;

    threadpool_lock(pool);

    timer = &pool->thread_timer[tid - 1];
    if (timer->ref)
    {
        cv_kill = timer->cv_kill;
        if (timer->threadid && timer->threadid != get_current_thread_id())
        {
            //timer回调线程中删除定时器,那么用户应该知道如何保证线程安全
            //直接删除就可以了, 如果非当前线程, 那么要等timer回调结束,
            //才能删除定时器
            threadpool_wait_cond(pool, timer->cv_kill, FOREVER);
        }

        min_heap_erase_(pool->time_heap, timer->index);

        empty(timer, sizeof(POOLTIMER));
        timer->cv_kill = cv_kill; //还原一下,这个不能清除

        pool->timer_cur_count--;

        ret = TRUE;
    }

    threadpool_unlock(pool);

    return ret;
}


/*
拓展的队列接口
*/
HQUEUE queue_create(unsigned int max_message)
{
    THREADPOOL_CONFIG config = {0};

    //以下提供一些默认值，用户可以在threadpool_create前将其修改掉
    config.thread_num = 0;
    config.max_thread_num = 0;
    config.queue_max_num = max_message; //队列可容纳max_message个任务
    config.max_timer_count = 64; //最大可创建32个定时器
    config.priority_count = 1; //只有一个优先级

    return (HQUEUE)threadpool_create(&config);
}

unsigned int queue_dispatch(HQUEUE queue)
{
    unsigned int ret = threadpool_worker(queue);
    return ret;
}

int queue_post(HQUEUE queue, queue_hook_cb callback, void* obj, void* arg)
{
    return threadpool_add_job((THREADPOOL)queue, 0, TRUE, callback, obj, arg);
}

thread_bool_t queue_break(HQUEUE queue)
{
    ((THREADPOOL)queue)->pool_close = TRUE;
    return TRUE;
}

thread_bool_t queue_destroy(HQUEUE queue)
{
    return threadpool_destroy((THREADPOOL)queue);
}

HQTIMER queue_set_timer(HQUEUE queue, thread_ulong_t timeout, timer_hook_cb callback, void* arg, long event)
{
    return threadpool_set_timer((THREADPOOL)queue, timeout, callback, arg, event);
}

thread_bool_t queue_kill_timer(HQUEUE queue, HQTIMER timer)
{
    return threadpool_kill_timer((THREADPOOL)queue, timer);
}
